package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.bean.KeywordBean;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.IkAnalyzer;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/2 15:17
 */
public class Dws_01_DwsTrafficSourceKeywordPageViewWindowApp extends BaseSQLApp {
    
    public static void main(String[] args) {
        new Dws_01_DwsTrafficSourceKeywordPageViewWindowApp().init(
            "DwsTrafficSourceKeywordPageViewWindowApp",
            4001,
            1,
            "DwsTrafficSourceKeywordPageViewWindowApp"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        /*
           1. 读取页面日志数据
           2. 过滤其中的搜索行为数据
                item is not null
                last_pge_id = search
                item_type = keyword
           3. 对内容进行分词
           
           4. 开窗聚合
                可以使用分组窗口
                
                也可以使用tvf
         */
        tEnv.executeSql(
            "create table page_log( " +
                "`common` map<string, string>, " +
                "`page` map<string, string>, " +
                "`ts` bigint, " +
                "row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000, 'yyyy-MM-dd HH:mm:ss')), " +
                "WATERMARK FOR row_time AS row_time - INTERVAL '3' SECOND " +
                ")" + SQLUtil.getKafkaDDL("DwsTrafficSourceKeywordPageViewWindowApp", Constant.TOPIC_DWD_TRAFFIC_PAGE));
        // TODO 4. 从表中过滤搜索行为
        Table searchTable = tEnv.sqlQuery(
            "select " +
                "page['item'] full_word, " +
                "row_time " +
                "from page_log " +
                "where page['item'] is not null " +
                "and page['last_page_id'] = 'search' " +
                "and page['item_type'] = 'keyword'");
        tEnv.createTemporaryView("search_table", searchTable);
        //tEnv.sqlQuery("select * from search_table").execute().print();
        
        // TODO 5. 使用自定义的UDTF函数对搜索的内容进行分词
        tEnv.createTemporaryFunction("ik_analyze", IkAnalyzer.class);
        Table splitTable = tEnv.sqlQuery(
            "select " +
                "keyword, " +
                "row_time  " +
                "from search_table, " +
                "lateral table(ik_analyze(full_word)) " +
                "as t(keyword)");
        tEnv.createTemporaryView("split_table", splitTable);
//
//        tEnv.sqlQuery("select * from  split_table").execute().print();
        
        // TODO 6. 分组、开窗、聚合计算
        Table KeywordBeanSearch = tEnv.sqlQuery(
            "select " +
                "DATE_FORMAT(TUMBLE_START(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt, " +
                "DATE_FORMAT(TUMBLE_END(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt, " +
                "'search' source, " +
                "keyword, " +
                "count(*) keyword_count, " +
                "UNIX_TIMESTAMP()*1000 ts " +
                "from split_table " +
                "GROUP BY TUMBLE(row_time, INTERVAL '10' SECOND),keyword");
        
        // TODO 7. 将动态表转换为流
        DataStream<KeywordBean> keywordBeanDS = tEnv.toAppendStream(KeywordBeanSearch, KeywordBean.class);
        
        // TODO 8. 将流中的数据写到ClickHouse中
        SinkFunction<KeywordBean> jdbcSink = FlinkSinkUtil
            .getClickHouseSink("gmall2022",
                               "dws_traffic_source_keyword_page_view_window",
                               KeywordBean.class
            );
        keywordBeanDS.addSink(jdbcSink);
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
